package cn.ac.iie.ulss.dataredistribution.handler;

import cn.ac.iie.ulss.dataredistribution.commons.GlobalVariables;
import cn.ac.iie.ulss.dataredistribution.commons.RuntimeEnv;
import cn.ac.iie.ulss.dataredistribution.tools.HNode;
import cn.ac.iie.ulss.dataredistribution.tools.MetaStoreClientPool;
import cn.ac.iie.ulss.dataredistribution.tools.INode;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author: evan
 * @date: 2014-10-15
 */
public class PrintEnvironment implements Runnable {

    String[] topics = null;
    ConcurrentLinkedQueue[] dataPool = null;
    Map<String, ConcurrentLinkedQueue[]> topicToDataPool = null;
    MetaStoreClientPool mscp = null;
    ConcurrentHashMap<String, ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>>> tableToMSGStation = null;
    static org.apache.log4j.Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = org.apache.log4j.Logger.getLogger(PrintEnvironment.class.getName());
    }

    @Override
    public void run() {
        init();
        StringBuffer datapoolsize = new StringBuffer();

        while (true) {
            for (String topic : topics) {
                dataPool = topicToDataPool.get(topic);
                if (dataPool == null) {
                    continue;
                } else {
                    datapoolsize.delete(0, datapoolsize.length());
                    for (int j = 0; j < dataPool.length; j++) {
                        datapoolsize.append(" ");
                        datapoolsize.append(dataPool[j].size());
                    }
                }

                logger.info(topic + " dataPool's size: " + datapoolsize + " metastoreclientpool'size is " + mscp.size());
            }

            try {
                Thread.sleep(10000);
            } catch (Exception ex) {
//                Logger.getLogger(PrintEnvironment.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    private void init() {
        topics = ((String) RuntimeEnv.getParam(RuntimeEnv.MQ_TOPIC_NAME)).split("\\|");
        topicToDataPool = (Map<String, ConcurrentLinkedQueue[]>) RuntimeEnv.getParam(GlobalVariables.TOPIC_TO_DATAPOOL);
        mscp = (MetaStoreClientPool) RuntimeEnv.getParam(GlobalVariables.METASTORE_CLIENT_POOL);
        tableToMSGStation = (ConcurrentHashMap<String, ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>>>) RuntimeEnv.getParam(GlobalVariables.TABLE_TO_MSGSTATION);
    }

    private int numOfQueues(String table) {
        int count = 0;
        ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>> msg = tableToMSGStation.get(table);
        for (HNode n : msg.keySet()) {
            ConcurrentHashMap<INode, ConcurrentLinkedQueue> chm = msg.get(n);
            count += chm.size();
        }
        return count;
    }
}
